package com.xc.quickstart.mq.listener;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.exceptions.ExceptionUtil;
import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.xc.quickstart.common.bean.ExceptionDelayQueueProcessor;
import com.xc.quickstart.common.context.ApiContext;
import com.xc.quickstart.common.event.ExceptionDelayQueueEvent;
import com.xc.quickstart.common.exception.CommonBizException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.Instant;
import java.util.Collections;
import java.util.Set;

/**
 * @author yinyansheng
 */
@Slf4j
public abstract class BaseMqListener implements MessageListener {

    @Autowired
    private ExceptionDelayQueueProcessor exceptionDelayQueueProcessor;

    @Override
    public Action consume(Message message, ConsumeContext context) {
        String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);
        try {
            if (CollectionUtil.isNotEmpty(getTags()) && !getTags().contains(message.getTag())) {
                log.warn("[消息消费][{}]不处理TAG不是[{}]的消息,message:{},messageBody:{},crmTraceId:{}"
                        , getListenerName(), JSON.toJSONString(getTags()), JSON.toJSONString(message), messageBody, ApiContext.getCrmTraceId());
                return Action.CommitMessage;
            }

            log.info("[消息消费][{}]开始,message:{},messageBody:{},crmTraceId:{}", getListenerName(), JSON.toJSONString(message), messageBody, ApiContext.getCrmTraceId());

            if (message.getReconsumeTimes() >= getMaxReconsumeTimes()) {
                log.error("[消息消费][{}]失败，已经重试{}次,message:{},messageBody:{},crmTraceId:{}"
                        , message.getReconsumeTimes(), getListenerName(), JSON.toJSONString(message), messageBody, ApiContext.getCrmTraceId());
                return Action.CommitMessage;
            }

            Instant start = Instant.now();
            Action action = this.doConsume(message, context);

            log.info("[消息消费][{}]结束,message:{},messageBody:{},result:{},耗时:{}ms,crmTraceId:{}"
                    , getListenerName()
                    , JSON.toJSONString(message)
                    , messageBody
                    , JSON.toJSONString(action)
                    , Duration.between(start, Instant.now()).toMillis()
                    , ApiContext.getCrmTraceId());

            return Action.CommitMessage;
        } catch (CommonBizException ex) {
            log.error("[消息消费][{}]CRM异常;message:{};messageBody:{};ex:{},crmTraceId:{}"
                    , getListenerName()
                    , JSON.toJSONString(message)
                    , messageBody
                    , ExceptionUtil.stacktraceToString(ex)
                    , ApiContext.getCrmTraceId());

            exceptionDelayQueueProcessor.put(new ExceptionDelayQueueEvent(this.getClass().getName(), getListenerName(), ExceptionUtil.stacktraceToString(ex, 400)));
            return Action.CommitMessage;
        } catch (Exception ex) {
            log.error("[消息消费][{}]异常;message:{};messageBody:{};ex:{},crmTraceId:{}"
                    , getListenerName()
                    , JSON.toJSONString(message)
                    , messageBody
                    , ExceptionUtil.stacktraceToString(ex)
                    , ApiContext.getCrmTraceId());

            exceptionDelayQueueProcessor.put(new ExceptionDelayQueueEvent(this.getClass().getName(), getListenerName(), ExceptionUtil.stacktraceToString(ex, 400)));
            return Action.ReconsumeLater;
        } finally {
            ApiContext.removeApiContextModel();
        }
    }

    public abstract String getGroupId();

    public abstract String getTopic();

    protected abstract Action doConsume(Message message, ConsumeContext context);

    protected abstract String getListenerName();

    protected Integer getMaxReconsumeTimes() {
        return 3;
    }

    public String getConsumeThreadNums() {
        return "10";
    }

    public Set<String> getTags() {
        return Collections.emptySet();
    }

}
